package com.zxy.ziems.common.mqtt;

import com.zxy.btp.common.util.JsonUtils;
import com.zxy.ziems.common.mqtt.pojo.MqttMessagePayload;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.core.GenericTypeResolver;

import java.time.LocalDateTime;
import java.util.Objects;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * mqttConsumer基类
 * @author 10348359
 * @create 2024-04-29 14:58
 */
@Slf4j
public abstract class BaseMqttConsumer<T> implements IMqttMessageListener, MqttMessageDecoder<MqttMessagePayload<T>> {
    private Class<T> entityClass;

    public BaseMqttConsumer() {
        Class<?>[] typeArguments = GenericTypeResolver.resolveTypeArguments(this.getClass(), BaseMqttConsumer.class);
        this.entityClass = (Class<T>) typeArguments[0];
    }

    public static ExecutorService executorService = Executors.newFixedThreadPool(10);
    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) {
        log.debug("[MQTT]\r\n 收到主题 :\r\n" + topic + " 的消息:\r\n" + new String(mqttMessage.getPayload()));
        executorService.submit(() -> {
            try {
                MqttMessagePayload<T> decoder = mqttMessageDecode(mqttMessage);
                T entity;
                LocalDateTime sendTime;
                if (Objects.nonNull(decoder)) {
                    entity = decoder.getData();
                    sendTime = decoder.getTime();
                } else {
                    entity = null;
                    sendTime = null;
                }
                doProcessMessage(topic, entity, sendTime);
            } catch (Exception ex) {
                log.error(ex.toString());
            }
        });
    }

    /**
     * 业务操作
     * @param topic
     * @param entity
     */
    protected abstract void doProcessMessage(String topic, T entity, LocalDateTime sendTime);

    @Override
    public MqttMessagePayload<T> mqttMessageDecode(MqttMessage msg) {
        return JsonUtils.parseObject(new String(msg.getPayload()), MqttMessagePayload.class, entityClass);
    }
}

